package pageRank;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.Counters.Counter;

public class Calc extends Job {

	private int round;
	private String input;
	private String output;
	
	public Calc(String input, String output, int round, String... args) {
		super("PageRank " + round, args);
		this.round = round;
		this.input = input;
		this.output = output;
	}

	@Override
	protected CombineConfig getCombineConfig() {
		return CombineConfig.newInstance(PageRank_Combiner.class);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		return InputConfig.newInstance(SequenceFileInputFormat.class,
				input+"/PageRank_" + (round - 1));
	}

	@Override
	protected String getLogPath() {
		return "/user/jwLiu/CJKPageRank/logs";
	}

	@Override
	protected MapConfig getMapConfig() {
		return MapConfig.newInstance(Calc_Mapper.class, 0, BytesWritable.class,
				BytesWritable.class, DefaultCodec.class);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		return OutputConfig.newInstance(SequenceFileOutputFormat.class, BytesWritable.class,
				BytesWritable.class, output+"/PageRank_" + round, GzipCodec.class);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		return ReduceConfig.newInstance(PageRank_Reducer.class, 2000);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
		SequenceFileOutputFormat.setOutputCompressionType(conf, SequenceFile.CompressionType.BLOCK);
		conf.set("Round", String.valueOf(round));
		conf.set("dfs.http.address", "10.251.2.1:50070");
		conf.set("io.sort.mb", "300");
		conf.set("mapred.child.java.opts", "-Xmx5g");
		
		if (round == 1) {
			conf.set("Z", "0");
		} else {
			double z = Double.parseDouble(new BufferedReader(new InputStreamReader(
					fs.open(new Path("Z_" + (round - 1))))).readLine());
			long websize = Long.parseLong(new BufferedReader(new InputStreamReader(
					fs.open(new Path("WebSize_" + (round - 1))))).readLine());
			conf.set("Z", String.valueOf(z * 0.85 / websize));
		}
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
		long websize = job.getCounters().findCounter("PageRank", "WebSize")
				.getCounter();
		BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs
				.create(new Path("WebSize_" + round), true)));
		bw.write(String.valueOf(websize));
		bw.flush();
		bw.close();
		
		Counter delta = job.getCounters().findCounter("PageRank", "Delta");
		if (delta == null || delta.getCounter() == 0) {
			try {
				new Readable(output,"/user/jwLiu/PageRank/Exception",round, new String[0]).submit();
			} catch (Exception e) {
				throw new IOException(e);
			}
		}
	}

}
